package com.atguigu.flink.sql.query;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by 黄凯 on 2023/6/26 0026 18:19
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 基本查询演示
 */
public class Flink01_BaseQuery {

    public static void main(String[] args) {

        //准备表环境
        TableEnvironment tableEnv = TableEnvironment.create(EnvironmentSettings.newInstance().build());

        //kafka Connetor
        String readSql =
                " create table t1 (" +
                        " id STRING , " +
                        " vc INT , " +
                        " ts BIGINT, " +
                        " pt AS PROCTIME(), " +
                        " et AS TO_TIMESTAMP_LTZ(ts, 3) , " +
                        " watermark for et as et - interval '1' second " +
                        ") WITH (" +
                        " 'connector' = 'kafka', " +
                        " 'topic' = 'topicA', " +
                        " 'properties.bootstrap.servers' = 'hadoop102:9092', " +
                        " 'properties.group.id' = 'flinksql', " +
                        " 'scan.startup.mode' = 'latest-offset', " +
                        " 'value.format' = 'csv' " +
                        ")";
        tableEnv.executeSql(readSql);
        Table table = tableEnv.from("t1");
        table.printSchema();

        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        // 从Socket读取数据
        SingleOutputStreamOperator<WaterSensor> stream = env.socketTextStream("hadoop102", 8888)
                .map(
                        value -> {
                            String[] fields = value.split(",");
                            return new WaterSensor(fields[0].trim(), Long.valueOf(fields[2].trim()), Integer.valueOf(fields[1].trim()));
                        }
                );

        // 流转表
        Schema schema =
                Schema.newBuilder()
                        .column("id", "STRING")
                        //.column("user", DataTypes.STRING())
                        .column("vc", "INT")
                        .column("ts", "BIGINT")
                        //处理时间字段
                        .columnByExpression("pt", "proctime()")
                        //事件时间字段 , 基于表中的某个列计算得到。
                        .columnByExpression("et", "TO_TIMESTAMP_LTZ(ts , 3)")
                        .watermark("et", "et - INTERVAL '1' SECOND")
                        .build();
        Table table1 = tableEnvironment.fromDataStream(stream, schema);
        tableEnvironment.createTemporaryView("t2", table1);

        // select where
        //tableEnv.sqlQuery("select id , vc , ts  from t1 where vc >= 100 ").execute().print();

        //with
        //tableEnv.sqlQuery("with tmp as (select id , vc , ts  from t1 where vc >= 100) select * from tmp").execute().print();

        //distinct
        //tableEnv.sqlQuery("select distinct id from t1 ").execute().print();

        //group by
        //tableEnv.sqlQuery("select id ,count(vc) from t1 group by id ").execute().print();

        //分组聚合的多维度分析
        /*
          需求: 有一张表A, 其中有 user amount  a  b  c 几个字段， a b c 是三个维度字段
               期望通过不同维度的组合进行sum(amount) , 最终的结果整体呈现。

            笨办法： union
               select sum(amount)  from A
               union
               select a , sum(amount)  from A group by a
               union
               select b , sum(amount) from A  group by b
               union
               select c , sum(amount) from A group by c
               union
               select a, b , sum(amount) from A group by a,b
               union
               select a, c , sum(amount) from A group by a ,c
               union
               select b,c , sum(amount) from A group by b, c
               union
               select a, b, c , sum(amount) from A group by a, b , c

            好办法: grouping sets
               grouping sets:
                 select a , b , c , sum(amount ) group by grouping sets ( () ,a ,b , c ,(a,b) , (a,c) , (b,c) , (a,b,c))
               cube:
                 select a, b , c ,sum(amount) GROUP BY CUBE (a, b, c )
               rollup:
                 select a, b ,c  ,sum(amount) group by ROLLUP (a, b, c )


           Hive Grouping sets :
               grouping sets :
                   SELECT a, b, SUM( c ) FROM tab1 GROUP BY a, b GROUPING SETS ( (a, b), a, b, ( ) )
               rollup:
                   SELECT a, b, SUM( c ) FROM tab1  GROUP BY a, b, c WITH ROLLUP
               cube:
                 SELECT a, b, SUM( c ) FROM tab1  GROUP BY a, b, c WITH CUBE

         */

        // grouping sets
        /**
         * +----+--------------------------------+-------------+----------------------+
         * | op |                             id |          ts |                  cnt |
         * +----+--------------------------------+-------------+----------------------+
         * | +I |                            ws3 |      <NULL> |                    1 |
         * | +I |                         <NULL> |        5000 |                    1 |
         * | +I |                            ws1 |        3000 |                    1 |
         * | +I |                            ws4 |        6000 |                    1 |
         * | +I |                            ws1 |        1000 |                    1 |
         * | +I |                         <NULL> |        4000 |                    1 |
         * | +I |                            ws2 |      <NULL> |                    1 |
         * | +I |                         <NULL> |        3000 |                    1 |
         * | -U |                            ws2 |      <NULL> |                    1 |
         * | +U |                            ws2 |      <NULL> |                    2 |
         * | +I |                            ws1 |      <NULL> |                    1 |
         * | -U |                            ws1 |      <NULL> |                    1 |
         * | +U |                            ws1 |      <NULL> |                    2 |
         * | +I |                            ws2 |        4000 |                    1 |
         * | +I |                            ws3 |        5000 |                    1 |
         * | +I |                         <NULL> |        1000 |                    1 |
         * | +I |                            ws2 |        1000 |                    1 |
         * | -U |                         <NULL> |        1000 |                    1 |
         * | +U |                         <NULL> |        1000 |                    2 |
         * | +I |                            ws4 |      <NULL> |                    1 |
         * | +I |                         <NULL> |        6000 |                    1 |
         * +----+--------------------------------+-------------+----------------------+
         */
        /*tableEnv.sqlQuery("select id , ts  , count(*) cnt from \n" +
                "                ( VALUES \n" +
                "                        ('ws1' , 1000 , 1 ) ,\n" +
                "                        ('ws2' , 1000 , 2 ) ,\n" +
                "                        ('ws1' , 3000 , 3 ) ,\n" +
                "                        ('ws2' , 4000 , 4 ) ,\n" +
                "                        ('ws3' , 5000 , 2 ) ,\n" +
                "                        ('ws4' , 6000 , 5 )) \n" +
                "                AS ws (id , ts , vc )\n" +
                "                group by GROUPING SETS (\n" +
                "                        (id,ts) ,\n" +
                "                        (id) ,\n" +
                "                        (ts)\n" +
                "                ) \n").execute().print();*/

        //rollup
        /**
         * +----+--------------------------------+-------------+----------------------+
         * | +I |                            ws3 |      <NULL> |                    1 |
         * | +I |                            ws2 |        1000 |                    1 |
         * | +I |                            ws4 |      <NULL> |                    1 |
         * | +I |                            ws3 |        5000 |                    1 |
         * | +I |                            ws1 |        1000 |                    1 |
         * | +I |                            ws1 |      <NULL> |                    1 |
         * | -U |                            ws1 |      <NULL> |                    1 |
         * | +U |                            ws1 |      <NULL> |                    2 |
         * | +I |                            ws2 |        4000 |                    1 |
         * | +I |                         <NULL> |      <NULL> |                    1 |
         * | -U |                         <NULL> |      <NULL> |                    1 |
         * | +U |                         <NULL> |      <NULL> |                    2 |
         * | +I |                            ws1 |        3000 |                    1 |
         * | -U |                         <NULL> |      <NULL> |                    2 |
         * | +U |                         <NULL> |      <NULL> |                    3 |
         * | -U |                         <NULL> |      <NULL> |                    3 |
         * | +U |                         <NULL> |      <NULL> |                    4 |
         * | -U |                         <NULL> |      <NULL> |                    4 |
         * | +U |                         <NULL> |      <NULL> |                    5 |
         * | +I |                            ws4 |        6000 |                    1 |
         * | -U |                         <NULL> |      <NULL> |                    5 |
         * | +U |                         <NULL> |      <NULL> |                    6 |
         * | +I |                            ws2 |      <NULL> |                    1 |
         * | -U |                            ws2 |      <NULL> |                    1 |
         * | +U |                            ws2 |      <NULL> |                    2 |
         * +----+--------------------------------+-------------+----------------------+
         */
        /*tableEnv.sqlQuery(
                "select id , ts  , count(*) cnt from \n" +
                        "                ( VALUES \n" +
                        "                        ('ws1' , 1000 , 1 ) ,\n" +
                        "                        ('ws2' , 1000 , 2 ) ,\n" +
                        "                        ('ws1' , 3000 , 3 ) ,\n" +
                        "                        ('ws2' , 4000 , 4 ) ,\n" +
                        "                        ('ws3' , 5000 , 2 ) ,\n" +
                        "                        ('ws4' , 6000 , 5 )) \n" +
                        "                AS ws (id , ts , vc )\n" +
                        "                group by ROLLUP(\n" +
                        "                 id ,ts\n" +
                        "                )\n").execute().print();*/


        // cube
        /**
         * +----+--------------------------------+-------------+----------------------+
         * | +I |                            ws1 |        1000 |                    1 |
         * | +I |                         <NULL> |        4000 |                    1 |
         * | +I |                            ws3 |        5000 |                    1 |
         * | +I |                         <NULL> |        1000 |                    1 |
         * | +I |                            ws2 |        1000 |                    1 |
         * | -U |                         <NULL> |        1000 |                    1 |
         * | +U |                         <NULL> |        1000 |                    2 |
         * | +I |                            ws4 |      <NULL> |                    1 |
         * | +I |                         <NULL> |        6000 |                    1 |
         * | +I |                            ws1 |      <NULL> |                    1 |
         * | -U |                            ws1 |      <NULL> |                    1 |
         * | +U |                            ws1 |      <NULL> |                    2 |
         * | +I |                            ws2 |        4000 |                    1 |
         * | +I |                            ws2 |      <NULL> |                    1 |
         * | +I |                         <NULL> |        3000 |                    1 |
         * | -U |                            ws2 |      <NULL> |                    1 |
         * | +U |                            ws2 |      <NULL> |                    2 |
         * | +I |                         <NULL> |      <NULL> |                    1 |
         * | -U |                         <NULL> |      <NULL> |                    1 |
         * | +U |                         <NULL> |      <NULL> |                    2 |
         * | +I |                            ws1 |        3000 |                    1 |
         * | -U |                         <NULL> |      <NULL> |                    2 |
         * | +U |                         <NULL> |      <NULL> |                    3 |
         * | -U |                         <NULL> |      <NULL> |                    3 |
         * | +U |                         <NULL> |      <NULL> |                    4 |
         * | -U |                         <NULL> |      <NULL> |                    4 |
         * | +U |                         <NULL> |      <NULL> |                    5 |
         * | +I |                            ws4 |        6000 |                    1 |
         * | -U |                         <NULL> |      <NULL> |                    5 |
         * | +U |                         <NULL> |      <NULL> |                    6 |
         * | +I |                            ws3 |      <NULL> |                    1 |
         * | +I |                         <NULL> |        5000 |                    1 |
         * +----+--------------------------------+-------------+----------------------+
         */
        tableEnv.sqlQuery("select id , ts  , count(*) cnt from \n" +
                "                ( VALUES \n" +
                "                        ('ws1' , 1000 , 1 ) ,\n" +
                "                        ('ws2' , 1000 , 2 ) ,\n" +
                "                        ('ws1' , 3000 , 3 ) ,\n" +
                "                        ('ws2' , 4000 , 4 ) ,\n" +
                "                        ('ws3' , 5000 , 2 ) ,\n" +
                "                        ('ws4' , 6000 , 5 )) \n" +
                "                AS ws (id , ts , vc )\n" +
                "                group by CUBE(\n" +
                "                  id ,ts\n" +
                "                )\n").execute().print();

        //order by
        //tableEnv.sqlQuery("select id , vc , ts , et  from t1 order by et ").execute().print();

        //limit batch

        //sql hits
        //tableEnv.sqlQuery("select id, vc, ts  from t1 /*+ OPTIONS('topic'='topicB') */").execute().print();

        //集合操作
        tableEnv.executeSql("create view t3(s) as values ('c'), ('a'), ('b'), ('b'), ('c')") ;
        tableEnv.executeSql("create view t4(s) as values ('d'), ('e'), ('a'), ('b'), ('b')") ;

        // UNION  |  UNION ALL  并集
        //tableEnv.sqlQuery("(SELECT s FROM t3) UNION (SELECT s FROM t4)").execute().print();
        //tableEnv.sqlQuery("(SELECT s FROM t3) UNION ALL (SELECT s FROM t4)").execute().print();

        //INTERSECT | INTERSECT ALL  交集
        //tableEnv.sqlQuery("(SELECT s FROM t3) INTERSECT (SELECT s FROM t4)").execute().print();
        //tableEnv.sqlQuery("(SELECT s FROM t3) INTERSECT ALL (SELECT s FROM t4)").execute().print();


        //EXCEPT | EXCEPT ALL  差
        //tableEnv.sqlQuery("(SELECT s FROM t3) EXCEPT (SELECT s FROM t4)").execute().print();//c
        //tableEnv.sqlQuery("(SELECT s FROM t3) EXCEPT ALL (SELECT s FROM t4)").execute().print(); // c c


    }

}
